package com.yj.rocketmq;

import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
import org.apache.rocketmq.common.message.Message;
import org.springframework.stereotype.Component;
import org.springframework.util.Assert;

import javax.annotation.PostConstruct;
import java.text.SimpleDateFormat;
import java.util.Date;

/**
 * RocketMQ工具类
 */
@Slf4j
@Component
public class RocketMqHandler {
    private static final SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
    /**
     * 生产者
     */
    private DefaultMQProducer producer;
    /**
     * 生产组名称
     */
    private String producerGroup = "producer-test";
    /**
     * RocketMQ Namesrv的地址
     */
    private String mqAddr = "127.0.0.1:9876";

    /**
     * 初始化生产者
     */
    @PostConstruct
    private void initProducer() {
        producer = new DefaultMQProducer(producerGroup);
        producer.setNamesrvAddr(mqAddr);
        try {
            // 在发送消息前启动Producer
            producer.start();
            log.info("producer start success. start time:{}", sdf.format(new Date()));
        } catch (MQClientException e) {
            log.error("start producer failed. error message:{}", e.getErrorMessage());
        }
    }

    /**
     * 发送消息
     * @param topic topic
     * @param tag   tag
     * @param msg   发送的消息
     */
    public SendResult sendMsg(String topic, String tag, String msg) {
        Assert.hasText(topic, "topic not blank");
        Assert.hasText(tag, "tag not blank");
        Assert.hasText(msg, "msg not blank");
        Message message = new Message(topic, tag, msg.getBytes());
        try {
            return producer.send(message);
        } catch (Exception e) {
            //***发送消息时会检查磁盘空间，已用90%就会发送失败，本地测试时由于C盘空间不足导致一直发送失败***
            log.error("send message failed. error message:{}", e.getMessage());
            return null;
        }
    }
}